CSEG8003 Course home Portal
UPES · School of Computer Science
CSEG8003 — Modelling and Simulation · L-T-P-C 2-0-1-3
Unit III
Converting to Parallel and Distributed Simulations
7 lecture hours · Theory notes · Dr. Mohsin Furkh Dar
CO2CO3 Partitioning Dependencies Load balancing Scalability Fault tolerance
The one idea behind this unit

Parallelising a simulation is not a coding problem; it is a decomposition problem. You are choosing where to cut the model so that the pieces are equal in work and the cuts are cheap in communication. Every topic in this unit — data partitioning, algorithm partitioning, dependency handling, dynamic repartitioning, graph partitioning, fault tolerance — is a consequence of that one trade-off:

minimise   maxp(workp)   subject to   minimal edge cut (communication)

Say this sentence in the first line of any long answer in this unit.

0. Why Parallelise a Simulation?

  1. Capability — the model does not fit in one machine's memory (a 1010-cell mesh, a 108-agent population).
  2. Capacity / turnaround — the run must finish before the decision is needed (a weather forecast for tomorrow is useless if it takes 30 hours).
  3. Statistical demand — the 1/√n law of Unit I means accurate stochastic results need thousands of replications.
  4. Geographic distribution — federated simulations (HLA) join simulators owned by different organisations that cannot be co-located.
  5. Hardware reality — single-core clock speed stopped increasing around 2005; all further performance is parallel.
Definition — Parallel vs. distributed simulation

Parallel simulation executes one model on multiple tightly coupled processors (shared memory or a fast interconnect) with the goal of reducing execution time.

Distributed simulation executes interacting simulators on geographically separated, loosely coupled machines, usually for interoperability and resource sharing as much as for speed.

0.1 The limits: Amdahl and Gustafson

Let f be the serial fraction of the work and p the number of processors.

Amdahl (fixed problem size):   S(p) = 1 / (f + (1−f )/p)  →  1/f as p→∞
Gustafson (fixed time, scaled problem):   S(p) = pf(p − 1)

With only 5% serial code, no machine on earth can exceed a 20× speedup on a fixed problem — that is Amdahl's warning. Gustafson's answer is that in practice we do not keep the problem fixed: given more processors we simulate a finer mesh or a bigger population, and the serial fraction shrinks relative to the growing parallel work. Hence the two notions of scaling in Section 6.

1. Partitioning the Data

Definition — Data (domain) decomposition

Data partitioning divides the model's state — the mesh, the grid, the particle set, the agent population, the graph — among processors, each of which applies the same algorithm to its own share (SPMD: single program, multiple data).

This is the dominant strategy in scientific simulation because it scales with problem size, whereas the number of distinct algorithmic stages is fixed.

1.1 Geometric decompositions of a mesh

Table 3.1 — Decomposing a structured N×N grid over p processors.
Scheme Shape of each partition Communication volume per processor
1-D (slab / stripe) N × N/p 2N — independent of p; simple, but surface-to-volume ratio worsens quickly
2-D (block / checkerboard) N/√p × N/√p 4N/√p — falls as p grows; the standard choice
3-D (cubic blocks) (N/p1/3)3 6N2/p2/3 — best surface-to-volume for 3-D problems
Recursive coordinate bisection Irregular boxes of equal work Good for non-uniform particle densities; cheap to compute
Space-filling curve (Morton/Hilbert) Contiguous runs of a linearised curve Preserves locality, extremely cheap to recompute — the standard for adaptive and particle codes

The surface-to-volume principle. Computation is proportional to the volume of a partition; communication is proportional to its surface. Always cut so that partitions are compact (low surface per unit volume). This single principle explains why 2-D beats 1-D and 3-D beats 2-D, and it is worth a diagram in the exam.

1.2 Halo (ghost) regions

A stencil computation needs the values of neighbouring cells that live on another processor. Each partition therefore keeps a halo — a read-only copy of the boundary layer of its neighbours — that is refreshed by a halo exchange once per step. Halo width equals the stencil radius; a wider stencil or a higher-order method costs proportionally more communication.

for each time step:
    start_halo_exchange()          # non-blocking Isend/Irecv
    compute_interior()             # overlap communication with computation
    wait_for_halo()
    compute_boundary()
    swap(old, new)

1.3 Partitioning other data structures

Common mistake

Splitting a random number stream badly. If every processor seeds its generator with, say, the rank, the streams may overlap and the replications are then not independent, silently invalidating every confidence interval. Use a generator with guaranteed independent substreams (e.g. counter-based generators, or MRG32k3a's stream/substream facility).

2. Partitioning the Algorithms

Definition — Functional (task) decomposition

Algorithm partitioning divides the work to be done rather than the data: different processors execute different functions, stages or model components, possibly on the same data.

2.1 Forms of algorithmic parallelism

  1. Pipelining. Stages arranged in sequence, each working on a different time step or dataset. Throughput rises by the number of stages; latency does not. Efficiency requires balanced stage times, and there is a fill/drain overhead.
  2. Component (model) parallelism. In a coupled climate model, the atmosphere, ocean, land and ice components run on separate processor groups and exchange fluxes at coupling intervals.
  3. Task parallelism with a task graph. Work is expressed as a DAG of tasks with dependencies; a runtime (Charm++, StarPU, Dask, Ray) schedules ready tasks onto free workers. This handles irregular workloads far better than static partitioning.
  4. Parameter sweep / replication parallelism. Independent runs across the scenario space — a task-parallel pattern with zero communication.
  5. Solver-internal parallelism. Parallel linear algebra, parallel FFT, parallel sorting inside a single time step.
  6. Speculative execution. Compute a branch before knowing whether it is needed; the basis of optimistic PDES (Section 3.4).

2.2 Data vs. algorithm partitioning

Table 3.2 — Comparing the two decomposition strategies.
Criterion Data partitioning Algorithm partitioning
What is split The state (mesh, particles, agents) The functions / stages / components
Code on each processor Same program (SPMD) Different programs (MPMD)
Scalability Scales with problem size — can use thousands of ranks Limited by the number of distinct stages/components
Load balance Good if the data is homogeneous; needs repartitioning otherwise Hard — stages rarely take equal time
Communication Boundary/halo exchange, mostly nearest-neighbour Stage-to-stage transfer of whole datasets
Typical use CFD, weather, MD, large ABMs Coupled multiphysics, real-time pipelines, workflow systems

Real large-scale codes use both: components are assigned to processor groups (algorithmic), and each group decomposes its own domain (data). That is the hybrid partitioning of Section 8.

3. Handling Inter-Partition Dependencies

Cutting a model creates dependencies across the cut. Managing them correctly — without destroying performance — is the central technical content of parallel simulation.

3.1 Kinds of dependency

3.2 Synchronous (time-stepped) execution

The easy case. All partitions execute step n, exchange haloes, hit a barrier , and proceed to step n+1. Correctness is automatic; the cost is that every barrier runs at the speed of the slowest processor, so load imbalance and OS jitter accumulate.

Definition — The causality constraint

In parallel discrete-event simulation, each logical process (LP) must process the events it receives in non-decreasing timestamp order. Violating this constraint — processing an event at t = 20 and then receiving one at t = 15 (a straggler) — produces results the sequential simulation would never produce.

3.3 Conservative synchronisation

An LP processes an event only when it can prove that no earlier event can still arrive.

  1. Each incoming link carries a clock: the timestamp of the last message received on it.
  2. The LP may safely process any event with timestamp less than the minimum over all input links.
  3. If a link is silent the LP blocks → potential deadlock.
  4. Chandy–Misra–Bryant null messages break the deadlock: an LP that will send nothing before time t sends a null message with timestamp t ("nothing from me until t").
  5. Performance depends critically on lookahead — the guaranteed minimum delay before an LP can affect another. Zero lookahead means no parallelism at all.

3.4 Optimistic synchronisation (Time Warp)

  1. Each LP processes events as fast as it can, assuming no straggler will arrive.
  2. State is saved (checkpointed) periodically so that it can be restored.
  3. When a straggler with timestamp ts arrives, the LP rolls back to the last state before ts.
  4. Messages sent in error are cancelled by anti-messages, which may cause cascading rollbacks in other LPs.
  5. Global Virtual Time (GVT) — the minimum timestamp of any unprocessed event or message in flight — is computed periodically. Nothing before GVT can ever be rolled back, so memory for older checkpoints is reclaimed (fossil collection) and irrevocable actions such as I/O are committed only up to GVT.
Table 3.3 — Conservative versus optimistic PDES.
Aspect Conservative (CMB) Optimistic (Time Warp)
Principle Never violate causality Violate, detect, and repair
Needs Good lookahead and known topology State saving and message cancellation
Overheads Blocking, null-message traffic Memory for checkpoints, rollback cost, anti-messages
Risk Deadlock; low parallelism when lookahead is poor Rollback thrashing / cascading rollbacks
Good for Models with real physical delays (networks with link latency, logistics) Irregular models with little exploitable lookahead
Exam tip

“Explain the causality problem in parallel simulation and its solutions” is the most predictable 10-mark question of this unit. Structure: definition of LP and causality constraint → the straggler example with numbers → conservative approach with lookahead and null messages → optimistic approach with rollback, anti-messages and GVT → comparison table → one sentence on when each is chosen.

3.5 Reducing dependency cost

4. Dynamic Partitioning and Load Balancing

Definition

Dynamic partitioning (dynamic load balancing) changes the assignment of data or tasks to processors during execution, in response to a workload that shifts over simulated time.

4.1 Why the load moves

4.2 The four questions of any load-balancing scheme

  1. Measure — what is the load metric? Cells, particles, events processed, or measured wall-clock time per rank (usually the most honest).
  2. Decide when — every k steps, or when imbalance λ = max/mean exceeds a threshold. Balance the benefit of rebalancing against its cost.
  3. Decide how — diffusive (shift work to lighter neighbours; small data movement, slow convergence) or global repartitioning (recompute the whole partition; better quality, more movement).
  4. Migrate — serialise the state, transfer, rebuild indices and neighbour lists, resume. Migration must preserve correctness of in-flight messages.

4.3 Techniques

Common mistake

Rebalancing too often. Each rebalance costs measurement, decision, data movement and index rebuilding. If the imbalance costs 3% and rebalancing costs 8%, you have made the code slower. The correct rule is to rebalance when the accumulated projected imbalance since the last rebalance exceeds the migration cost.

5. Communication Patterns in Partitioned Systems

Simple cost model:   Tcomm = α + nβ    (α = latency per message, β = time per byte, n = bytes)

On a modern cluster α ≈ 1–5 µs while 1/β is tens of GB/s. Sending 1000 messages of 8 bytes costs a thousand latencies; sending one message of 8000 bytes costs one. This is why message aggregation is the first optimisation to try.

5.1 The standard patterns

Table 3.4 — Communication patterns, their cost and where they occur.
Pattern Cost with p processors Occurs in
Point-to-point / nearest neighbour (halo) O(1) messages per rank; scales well Stencil codes, mesh simulations, spatial ABMs
Broadcast / scatter O(log p) with a tree Distributing parameters, initial conditions
Reduction / allreduce O(log p); a synchronisation point Global sums, convergence tests, adaptive Δt, GVT computation
Gather / allgather O(p) data volume per rank Collecting output, global agent lists (avoid at scale)
All-to-all (transpose) O(p) messages per rank — the most expensive pattern Parallel FFT, spectral methods, redistribution during rebalancing
Publish/subscribe, interest management Depends on the interest graph, not on p HLA/DDS distributed simulation, large multi-agent worlds
Asynchronous one-sided (RMA) No matching receive; overlaps well Irregular access, dynamic work stealing, PGAS models

5.2 Interest management

In distributed simulation with many entities, sending every update to everyone is O(n2 ) and impossible beyond a few thousand entities. Interest management (HLA Data Distribution Management) filters updates so that a federate receives only what it cares about, using routing spaces: the world is divided into regions or grid cells, each entity publishes to the cells it occupies and subscribes to the cells it can perceive. This turns a global broadcast into localised multicast.

5.3 Practical rules

  1. Prefer non-blocking communication and overlap it with interior computation.
  2. Aggregate small messages; pack contiguous buffers rather than sending strided data.
  3. Remove unnecessary collectives — a global reduction every step is often needed only every tenth step.
  4. Keep the communication topology matched to the network topology where possible (rank reordering).
  5. Measure with a profiler (Score-P, TAU, mpiP, Nsight) before optimising; intuition about where the time goes is usually wrong.

6. Scalability Challenges in Partitioned Systems

Definition — Strong and weak scaling

Strong scaling: fixed total problem size, increasing p; ideal behaviour is runtime ∝ 1/p. Limited by Amdahl's law and by the shrinking computation-to-communication ratio.

Weak scaling: fixed problem size per processor, so the total problem grows with p; ideal behaviour is constant runtime. This is how large simulations are actually used.

Speedup S(p) = T1/Tp,    Efficiency E(p) = S(p)/p

6.1 The eight barriers to scalability

  1. Serial fraction — initialisation, I/O, global decisions (Amdahl).
  2. Load imbalance — the barrier runs at the slowest rank; even 5% imbalance caps efficiency near 95% and it compounds every step.
  3. Communication growth — in strong scaling the partition shrinks, so the surface-to-volume ratio worsens and communication eventually dominates.
  4. Synchronisation and jitter — the barrier amplifies rare OS or network hiccups across all ranks.
  5. Collective operations — O(log p) at best, and allreduce latency becomes the floor of the time step at very large p.
  6. Memory per node — replicated global structures (a full agent directory, a full mesh copy) do not shrink with p and eventually exhaust memory.
  7. I/O bottleneck — thousands of ranks writing checkpoints to one file system; solved by parallel I/O (MPI-IO, HDF5, ADIOS) and in-situ analysis (Unit V).
  8. Fault probability — with more components, mean time between failures falls; at extreme scale the run will be interrupted (Section 9).
Example — why strong scaling stops

A 2-D 4096×4096 grid on p ranks gives blocks of side 4096/√p. Computation per rank ∝ 40962/p; communication ∝ 4×4096/√p. Their ratio is ≈ 1024/√p. At p = 1024 the ratio is 32 (fine); at p = 106 it is 1 — the rank now spends as long communicating as computing, and adding processors stops helping. The cure is to grow the problem (weak scaling), not the machine.

6.2 Diagnosing scalability

7. Partitioning in Graph-Based Systems

Definition — The graph partitioning problem

Given G = (V, E) with vertex weights (computation) and edge weights (communication), divide V into p disjoint parts such that the parts have approximately equal total vertex weight (balance constraint) and the total weight of edges between different parts (edge cut) is minimised. The problem is NP-hard, so heuristics are used.

7.1 Why graphs are the hard case

A mesh has geometry, so a coordinate cut works. A social, citation or web graph has no useful geometry, has a power-law degree distribution, and has a small diameter — so every balanced cut separates many edges. This is why distributed graph processing is dominated by communication.

7.2 Methods

7.3 Vertex-cut versus edge-cut

For power-law graphs, cutting vertices beats cutting edges. In the vertex-cut model (PowerGraph/GraphX) a high-degree hub is replicated across the machines that hold its edges, and its state is reconciled by a small gather–apply–scatter step. Because a hub with a million edges cannot live on one machine without wrecking the balance, this reformulation was the key advance for real-world graph systems.

7.4 Programming models for graph simulation

Example — partitioning a contact network for a parallel epidemic model

Vertices = people (weight = simulation cost), edges = daily contacts (weight = interaction frequency). ParMETIS produces p balanced parts with minimal cut; the cut edges become the cross-partition infection messages exchanged each day. Because households and workplaces form dense clusters, a good partitioner keeps them intact and the message volume falls by an order of magnitude compared with random assignment — which is precisely the experiment worth reporting in a lab record.

8. Hybrid Partitioning Approaches

Definition

Hybrid partitioning combines more than one decomposition strategy, or more than one level of parallelism, in the same simulation — typically because the hardware itself is hierarchical (cluster → node → socket → core → accelerator).

8.1 Forms of hybridisation

  1. MPI + OpenMP + CUDA. MPI between nodes (distributed memory), threads within a node (shared memory), and GPU kernels for the inner loops. Fewer, larger MPI partitions means less surface area and fewer messages.
  2. Data + task hybrid. Domain decomposition for the mesh, plus a task runtime for the irregular parts (chemistry, particle physics, adaptivity).
  3. Component + domain hybrid. Coupled multiphysics: each component owns a processor group (algorithmic partitioning) and decomposes its own domain (data partitioning).
  4. Multi-constraint / multi-objective partitioning. Balance two loads at once — e.g. cell count and particle count in a particle-in-cell code — which single weighted partitioning cannot do.
  5. Static + dynamic hybrid. A good static partition at start-up, with cheap diffusive corrections during the run.
  6. Mixed synchronisation. Conservative synchronisation between well-separated clusters with good lookahead, optimistic within a tightly coupled cluster.

8.2 Why hybrid usually wins

Common mistake

Assuming MPI+OpenMP is automatically faster than pure MPI. It is not: thread synchronisation, NUMA effects and a serial master thread during communication can eat the gain. Hybrid pays off when the halo volume is significant, when memory per rank is tight, or when the node count is very large. Always justify a hybrid design with a measurement.

9. Fault Tolerance in Partitioned Systems

At 105 nodes, even a per-node MTBF of ten years gives a system failure roughly every hour. A simulation that runs for a day must therefore be able to survive failures.

9.1 Failure model

9.2 Checkpoint and restart

Definition

Checkpoint/restart periodically saves a globally consistent snapshot of the simulation state to stable storage; after a failure the run resumes from the last checkpoint, losing at most one interval of work.

Young/Daly optimal interval:   Topt ≈ √(2 C M) C = checkpoint cost, M = MTBF

Techniques that reduce C: incremental checkpoints (save only changed pages), multilevel checkpointing (node-local SSD → partner node → parallel file system, as in SCR/FTI), asynchronous checkpointing (write in the background), and compression. In an application-level checkpoint the model writes only the physical state it actually needs, which is usually far smaller than a system-level memory image — and it is portable across machines.

9.3 Beyond checkpointing

Exam tip

A frequent 5-mark question: “derive/state the optimal checkpoint interval and explain the trade-off”. Say: too frequent → checkpoint overhead dominates; too rare → lost work after failure dominates; the total is minimised near √(2CM). Add one line on multilevel checkpointing as the practical improvement.

10. Partitioning for Emerging Architectures

The right partition depends on the machine. As hardware diversifies, partitioning must become architecture-aware.

Table 3.5 — Architectures and what they demand of the partitioning strategy.
Architecture Characteristics Partitioning implication
Many-core CPU / NUMA node Dozens of cores, non-uniform memory access Partition to respect NUMA domains; pin threads; first-touch allocation
GPU Thousands of SIMT lanes, high bandwidth, limited memory, costly host transfers Large, regular, coalesced partitions; avoid divergent branches; keep data resident on the device
Heterogeneous CPU+GPU nodes Very different per-device throughput Unequal (weighted) partitions; give the regular kernel to the GPU and irregular work to the CPU
FPGA / dataflow accelerators Custom pipelines, deterministic latency Pipeline (algorithmic) partitioning; stream-oriented data layout
Cloud / elastic clusters Variable node count, noisy neighbours, preemptible instances Elastic, migratable over-decomposition; failure treated as normal; cost-aware scheduling
Edge / fog Low bandwidth to the centre, limited local power Partition by data locality and privacy; keep raw data local, send summaries
Near-memory / processing-in-memory Compute placed next to memory banks Partition by memory locality rather than by compute load
Quantum / neuromorphic (early) Special-purpose co-processors within a classical workflow Offload a well-defined sub-problem; the classical partition surrounds it

10.1 Cross-cutting trends

  1. Data movement, not arithmetic, is the cost. Moving a double across a node can cost more energy than a hundred floating-point operations, so partitioning is increasingly an energy-optimisation problem.
  2. Asynchrony over barriers. Task-based, dependency-driven runtimes tolerate heterogeneity and jitter better than bulk-synchronous code.
  3. Performance portability. Kokkos, RAJA, SYCL and OpenMP target-offload let one partitioned code run on CPU and multiple GPU vendors.
  4. In-situ analysis. Because I/O does not scale, analysis and visualisation move into the simulation itself — the bridge to Unit V.
  5. Surrogates and ML acceleration. Learned models replace expensive kernels, and the partitioning must then balance a mix of physics and inference work.

11. A Practical Conversion Recipe

If an examiner (or a project) asks “how would you convert this sequential simulation to a parallel one?”, answer with this sequence:

  1. Profile first. Find the hot loops and the memory footprint; do not parallelise what does not matter.
  2. Try replication parallelism first. If the goal is statistical accuracy, run independent replications — near-linear speedup for almost no work.
  3. Choose the decomposition. Data if the state is large and homogeneous; algorithmic if the model is a pipeline of distinct components; hybrid at scale.
  4. Identify dependencies. Spatial (halo), temporal (causality), global (reductions), resource contention.
  5. Choose a synchronisation scheme. Barrier per step for time-stepped models; conservative or optimistic for event-driven ones, chosen by available lookahead.
  6. Design the communication. Non-blocking, aggregated, overlapped; remove unnecessary collectives.
  7. Handle randomness. Independent substreams per partition, and ensure the result is independent of the processor count.
  8. Verify against the sequential run. Same seed, same answer (or the same statistics with a documented reason for bitwise differences).
  9. Measure strong and weak scaling, and attribute lost efficiency to imbalance, communication or serial work.
  10. Add resilience — checkpoints at the Young/Daly interval, and a restart path that has actually been tested.
Common mistake — the reproducibility trap

Floating-point addition is not associative, so a parallel reduction over p ranks gives a slightly different sum for different p. In a chaotic model that difference grows exponentially and the trajectories diverge. Either use deterministic (fixed-order or compensated) reductions when bitwise reproducibility is required, or state clearly that only statistical reproducibility is claimed.

12. Unit Summary

12.1 Key terms

SPMD/MPMD · domain decomposition · halo/ghost cells · surface-to-volume ratio · Amdahl and Gustafson · strong and weak scaling · logical process · causality constraint · straggler · lookahead · null message · Time Warp · anti-message · GVT · fossil collection · work stealing · over-decomposition · space-filling curve · edge cut · vertex cut · multilevel partitioning · BSP superstep · interest management · allreduce · checkpoint interval · ABFT · ULFM.

12.2 Practice questions

Short answer (2–3 marks each)

  1. State Amdahl's law and give the maximum speedup when the serial fraction is 10%.
  2. Differentiate strong scaling from weak scaling.
  3. What is a halo region and why is it needed?
  4. Define lookahead and explain its role in conservative synchronisation.
  5. What are anti-messages in Time Warp?
  6. Why is a vertex cut preferred to an edge cut for power-law graphs?
  7. Write the optimal checkpoint-interval formula and define its symbols.

Medium answer (5 marks each)

  1. Compare data partitioning and algorithm partitioning under at least five criteria.
  2. Explain the surface-to-volume principle and use it to compare 1-D, 2-D and 3-D decompositions of a grid.
  3. Describe the four decisions involved in dynamic load balancing and two techniques used in practice.
  4. List the standard communication patterns with their cost in p and one simulation use for each.
  5. Explain multilevel graph partitioning (coarsen – partition – uncoarsen/refine).
  6. Describe three fault-tolerance techniques for large parallel simulations other than plain checkpointing.

Long answer (10 marks each)

  1. Explain the causality problem in parallel discrete-event simulation and compare conservative and optimistic synchronisation in detail, with examples of when each is preferred.
  2. Discuss the scalability challenges of partitioned simulations, with an analytical example showing why strong scaling saturates, and describe how each challenge is mitigated.
  3. Given a large agent-based epidemic simulation on a contact network, describe in full how you would convert it to a parallel and distributed implementation: partitioning, dependencies, synchronisation, communication, load balancing, randomness, verification and resilience.
  4. Discuss partitioning strategies for emerging architectures (GPU, heterogeneous nodes, cloud, edge), explaining how the objective changes from balancing computation to minimising data movement.
  5. Explain hybrid partitioning approaches and justify, with reasons and counter-arguments, when MPI+OpenMP+GPU is preferable to a flat MPI decomposition.

12.3 Further reading

CSEG8003 Modelling and Simulation · Unit III student notes · Dr. Mohsin Furkh Dar · UPES